package com.zgc.zkinfo;

import com.zgc.cache.ConnectionInfoCache;
import com.zgc.cache.CuratorClientCache;
import com.zgc.constant.ZkConstant;
import com.zgc.util.CuratorZkUtil;
import javafx.concurrent.Task;
import org.apache.curator.framework.CuratorFramework;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;

import java.util.concurrent.TimeUnit;

public class ZkTask extends Task<CuratorFramework> {
    private final Logger logger = LoggerFactory.getLogger(getClass());

    private final ZkConnectionInfo connectClient;
    private CuratorFramework zookeeperClient;

    public ZkTask(ZkConnectionInfo connectClient) {
        this.connectClient = connectClient;
    }

    public ZkConnectionInfo getZkConnectInfo() {
        return connectClient;
    }

    /*public void setZkConnectInfo(ZkConnectClient zkInfo) {
        this.connectClient = zkInfo;
    }*/


    /*@Override
    protected void running() {
        this.updateMessage("连接开始创建");
        this.updateProgress(1, 5);
        logger.debug("zk[{}]连接创建开始", connectClient.getHost());
    }*/

    /*@Override
    protected void scheduled() {
        super.updateProgress(2, 5);
    }*/

    @Override
    protected CuratorFramework call() throws Exception {
        super.updateMessage("连接创建中");
        logger.debug("zk[{}]连接创建中", connectClient.getHost());

        zookeeperClient = ZkCuratorClientFactory.builder(connectClient);
        zookeeperClient.start();

        if (!zookeeperClient.blockUntilConnected(connectClient.getConnectionWaitTimeout(), TimeUnit.MILLISECONDS)) {
            close();
            throw new RuntimeException("连接超时");
        }
        return zookeeperClient;
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        if (zookeeperClient != null) {
            connectClient.connectedProperty();
            close();
            logger.debug("zk[{}]连接关闭", connectClient.getHost());
        }
        refreshZkConnect(this.getValue());
        //this.updateProgress(0, 5);
        return super.cancel(mayInterruptIfRunning);
    }

    @Override
    protected void succeeded() {
        refreshZkConnect(this.getValue());
        CuratorCacheFactory.createTreeCacheListener(connectClient, ZkConstant.ROOT_NODE).start();
        this.updateMessage("连接成功");
        //this.updateProgress(5, 5);
        logger.debug("zk[{}]连接成功", connectClient.getHost());
    }

    @Override
    protected void failed() {
        refreshZkConnect(this.getValue());
        String message = "连接失败";
        Throwable exception = this.getException();
        if (exception != null && !StringUtils.isEmpty(exception.getMessage())) {
            message = exception.getMessage();
        }
        //this.updateProgress(-1, 5);
        this.updateMessage(message);
        logger.debug("zk[{}]{}", connectClient.getHost(), message);
    }

    // 连接成功后更新 CuratorFramework
    private void refreshZkConnect(CuratorFramework zookeeperClient) {
        if (zookeeperClient != null) {
            connectClient.setConnected(CuratorZkUtil.isConnect(zookeeperClient));
            CuratorClientCache.replaceCuratorClient(connectClient.getId(), this.getValue());
        }
        ConnectionInfoCache.replaceConnectInfo(connectClient.getId(), connectClient);
        this.updateValue(zookeeperClient);
    }

    private void close() {
        CuratorZkUtil.asynClose(zookeeperClient, connectClient);
        CuratorCacheFactory.close(connectClient.getId());
    }
}
